package com.atguigu.tingshu.search.mq;

import com.alibaba.cloud.commons.lang.StringUtils;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.search.repository.AlbumIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.vo.album.CanalListenAlbumStatMsgVo;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * Author  王东剑
 * Created on 12/6/2024 下午8:33
 * Package_name com.atguigu.tingshu.search.mq
 * Description
 */
@Slf4j
@Component
public class SearchReceiver {
	@Autowired
	private SearchService searchService;
	@Autowired
	private AlbumIndexRepository albumIndexRepository;

	@KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_UPPER)
	public void upperAlbum(ConsumerRecord<String,String> record){
		String albumId = record.value();
		if(!StringUtils.isEmpty(albumId))
			searchService.upperAlbum(Long.parseLong(albumId));
	}

	@KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_LOWER)
	public void lowerAlbum(ConsumerRecord<String,String> record){
		String albumId = record.value();
		if(!StringUtils.isEmpty(albumId))
			searchService.lowerAlbum(Long.parseLong(albumId));
	}

	@KafkaListener(topics = KafkaConstant.QUEUE_ALBUM_STAT_CANAL_TO_SEARCH)
	public void updateEsAlbumStatFromCanal(ConsumerRecord<String,String> record){
		String json = record.value();
		if(!StringUtils.isEmpty(json)){
			CanalListenAlbumStatMsgVo canalListenAlbumStatMsgVo = JSON.parseObject(json, CanalListenAlbumStatMsgVo.class);
			AlbumInfoIndex albumInfoIndex = albumIndexRepository.findById(canalListenAlbumStatMsgVo.getAlbumId()).get();
			switch (canalListenAlbumStatMsgVo.getStatType()){
				case SystemConstant.ALBUM_STAT_PLAY -> albumInfoIndex.setPlayStatNum(canalListenAlbumStatMsgVo.getStatNum());
				case SystemConstant.ALBUM_STAT_SUBSCRIBE -> albumInfoIndex.setSubscribeStatNum(canalListenAlbumStatMsgVo.getStatNum());
				case SystemConstant.ALBUM_STAT_BROWSE -> albumInfoIndex.setBuyStatNum(canalListenAlbumStatMsgVo.getStatNum());
				case SystemConstant.ALBUM_STAT_COMMENT -> albumInfoIndex.setCommentStatNum(canalListenAlbumStatMsgVo.getStatNum());
			}
			albumIndexRepository.save(albumInfoIndex);
			log.info("读取 Canal 发送至Kafka 的消息,并成功消费: {}",canalListenAlbumStatMsgVo);
		}

	}
}
